/*
 * Copyright 2011-2012 David Crosson
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

 
package fr.janalyse.series

import fr.janalyse.unittools._
import java.text.SimpleDateFormat


/**
  * TimeModel describes how cell's time is sampled
  * @author David Crosson
  */
case class TimeModel(val period:Long=1) {
    def referenceTime(time:Long) = time / period * period
    def granularityAt(time:Long) = period
    override def toString=""+period+"ms"
}

object TimeModel {
  implicit def long2TimeModel[P<%Number](period:P) = new TimeModel(period.longValue)
  implicit def string2TimeModel(desc:String) = new TimeModel(desc2Duration(desc))
}




/**
  * Series's cell root class, a cell is just a timestamp and a double value
  * @author David Crosson
  */
abstract class Cell {
  /**
    * time value, in most case will correspond to the java time, number of milliseconds ellapsed since seventies
    */
  def time:Long
  /**
    * the value of the cell
    */
  def value:Double
  /**
    * String representation of the cell
    */
  override def toString =  "(%s -> %.2f)".format(new SimpleDateFormat("yy-MM-dd HH:mm:ss").format(time), value)
}




/**
  * CalcCell is the cell type resulting of any mathematical operation
  * when combined, the new value replace the previous one
  */
case class CalcCell(time:Long, value:Double) extends Cell
/**
  * AddCell is used in order to add a new value to the current cell value
  * when combined, the new value is added to the previous one
  */
case class AddCell(time:Long, value:Double) extends Cell
/**
  * CountCell type is to be use when you're not really interested by the cell
  * value but by how many time you've combined a new cell to the current one
  */
case class CountCell(time:Long, value:Double) extends Cell
/**
  * StatCell saved statistics about concerning all values added to the same cell
  * Average, maximum, minimum, slop, ... 
  */
case class StatCell(time:Long, value:Double, count:Long, min:Double, max:Double, sum:Double) extends Cell {
  def this(time:Long, value:Double) = this(time, value, 1, value, value, value)
  def avg = sum/count
}
/**
  * QuoteCell store stock quote data  
  */
case class QuoteCell(time:Long, close:Double, open:Double, low:Double, high:Double, volume:Option[Double]) extends Cell {
  def this(time:Long, close:Double) = this(time, close, close, close, close, None)
  def this(time:Long, close:Double, volume:Double) = this(time, close, close, close, close, Some(volume))
  val value = close
  override def toString =  "(%s -> %.2f-%.2f <%.2f >%.2f #%d)".format(new SimpleDateFormat("yy-MM-dd HH:mm:ss").format(time), open, close, low, high, volume map {_.toLong} getOrElse 0L)
}




protected trait CellOrdering[C<:Cell] {
  implicit object defaultCellOrdering extends Ordering[C] {
    final def compare(c1: C, c2: C) = c1.value compare c2.value
  }  
}

object Cell       extends CellOrdering[Cell]
object CalcCell   extends CellOrdering[CalcCell]
object AddCell    extends CellOrdering[AddCell]
object CountCell  extends CellOrdering[CountCell]
object StatCell   extends CellOrdering[StatCell]
object QuoteCell  extends CellOrdering[QuoteCell]




trait CellBuilder[C <: Cell] {
  def buildFrom[T<%Number, V<%Number](time:T, value:V):C
  def buildFrom[X<:Cell](cell:X):C = buildFrom(cell.time, cell)
  def buildFrom[T<%Number, X<:Cell](time:T, cell:X):C = buildFrom(time, cell.value)
  def merge[X<:Cell](cellA:C, cellB:X):C
  //def buildFrom[T<%Number, A<%CellAnyContent](time:T, any:A):C = buildFrom(time, 1d)
  //def merge[X<:Cell, A<%CellAnyContent](cell:C, any:A):C = cell  // Default implementation = do nothing
}

object CellBuilder {
  implicit def calcCellBuilder = new CellBuilder[CalcCell] {
      def buildFrom[T<%Number, V<%Number](t:T, v:V) = CalcCell(t.longValue,v.doubleValue)
      def merge[X<:Cell](cellA:CalcCell, cellB:X) = CalcCell(cellA.time, cellB.value)
  }
  implicit def addCellBuilder = new CellBuilder[AddCell] {
      def buildFrom[T<%Number, V<%Number](t:T, v:V) = AddCell(t.longValue,v.doubleValue)
      def merge[X<:Cell](cellA:AddCell, cellB:X) = AddCell(cellA.time, cellA.value + cellB.value)
  }
  implicit def countCellBuilder = new CellBuilder[CountCell] {
      def buildFrom[T<%Number, V<%Number](t:T, v:V) = CountCell(t.longValue, 1)
      override def buildFrom[T<%Number, X<:Cell](t:T,cell:X) = cell match {
        case e:CountCell => CountCell(t.longValue, cell.value)
        case x => CountCell(t.longValue, 1)
      }
      def merge[X<:Cell](cellA:CountCell, cellB:X) = cellB match {
        case e:CountCell => CountCell(cellA.time, cellA.value+cellB.value)
        case x => CountCell(cellA.time, cellA.value+1)
      }
  }
  implicit def statCellBuilder = new CellBuilder[StatCell] {
      def buildFrom[T<%Number, V<%Number](t:T, v:V) = new StatCell(t.longValue,v.doubleValue)
      override def buildFrom[T<%Number, X<:Cell](t:T, cell:X) = cell match {
        case that:StatCell => StatCell(t.longValue,value=that.value, count=that.count, min=that.min, max=that.max, sum=that.sum)
        case that => new StatCell(t.longValue,that.value)
      }
      def merge[X<:Cell](cellA:StatCell, cellB:X):StatCell = cellB match {
        case that:StatCell =>
          StatCell(cellA.time,
                   (that.value+cellA.value*cellA.count)/(that.count+cellA.count),
                   that.count+cellA.count,
                   if (that.min < cellA.min) that.min else cellA.min,
                   if (that.max > cellA.max) that.max else cellA.max,
                   cellA.sum + that.sum)
        case that =>
          new StatCell(cellA.time,
                       (that.value+cellA.value*cellA.count)/(cellA.count+1),
                        cellA.count+1,
                        if (that.value < cellA.min) that.value else cellA.min,
                        if (that.value > cellA.max) that.value else cellA.max,
                        cellA.sum + that.value)
      }
    }
  
  implicit def quoteCellBuilder = new CellBuilder[QuoteCell] {
      def buildFrom[T<%Number, V<%Number](t:T, v:V):QuoteCell =  throw new RuntimeException("Not Allowed for QuoteCell") //new QuoteCell(t.longValue, v.doubleValue)
      //override def buildFrom[X<:Cell](cell:X):QuoteCell = buildFrom(cell.time, cell)
      override def buildFrom[T<%Number, X<:Cell](t:T, cell:X) = cell match {
        case that:QuoteCell =>
          QuoteCell(time   = t.longValue,
                    close  = that.close,
                    open   = that.open,
                    low    = that.low,
                    high   = that.high,
                    volume = that.volume)
        case that =>
          throw new RuntimeException("Couln't merge a QuoteCell with an ordinary cell...")
      }
      def merge[X<:Cell](cellA:QuoteCell, cellB:X):QuoteCell = cellB match {
        case that:QuoteCell =>
          QuoteCell(time   = cellA.time,
                    close  = if (that.time > cellA.time) that.close else cellA.close,
                    open   = if (that.time < cellA.time) that.open else cellA.open,
                    low    = if (that.low < cellA.low) that.low else cellA.low,
                    high   = if (that.high > cellA.high) that.high else cellA.high,
                    volume = for(s1 <-cellA.volume ; s2 <- that.volume) yield s1+s2 )
        case that =>
          throw new RuntimeException("Couln't merge a QuoteCell with an ordinary cell...")
      }
    }
}
// -----------------------------------------------------------------------------

/**
  * Series is a special kind of Cells collection. Cells are time ordered, and 
  * when two cells shares the same time (or time period) a combination operation
  * take place.
  * @author David Crosson
  */

class Series[+C <: Cell] (val name:String, val tm:TimeModel, val alias:String, protected val backend:IndexedSeq[C]=Vector.empty[C])
             (implicit builder:CellBuilder[C]) extends IndexedSeq[C] {
  /**
    * get series global statistics such as Average, Standard Deviation, ...
    * @return Series statistics
    */
  lazy val stat = Statistics(this)

  /**
    * Combine a new cell and the current series into a new series  
    * @param cell The new cell to combine
    * @return the resulting series
    */
  def << [X<:Cell] (cell:X) : Series[C] = new Series[C](name, tm, alias, combine(builder.buildFrom(cell)))

  /**
    * Combine a new number tuple and the current series into a new series  
    * @param cell the new cell, generated from the given tuple, to combine with
    * @return the resulting series
    */
  def << [T<%Number, V<%Number] (cell:Tuple2[T, V]):Series[C] = cell match {
    case (t,v) => new Series[C](name, tm, alias, combine(builder.buildFrom(t,v)))
  }

  private def combine[X>:C<:Cell](cell:X):IndexedSeq[X] = {
    val reftime=tm.referenceTime(cell.time)
    if (length==0 || reftime > last.time)  backend:+builder.buildFrom(reftime,cell)
    else if (reftime == last.time) backend.init:+builder.merge(backend.last, cell)
    else if (reftime < head.time) builder.buildFrom(reftime, cell)+:backend
    else {
      val index = binarySearchFirstGreaterOrEqual(backend, reftime)
      val (before, after) = backend.splitAt(index)
      // Deux cas : On remplace si égalité, on insère si absence.
      if (after.head.time==reftime) (before :+ builder.merge(after.head, cell)) ++ after.tail
      else  (before:+builder.buildFrom(reftime,cell)) ++ after
    }
  }


  // OK sauf pour ajout d'un element a la fin, mais ce cas n'arrive pas
  private def binarySearchFirstGreaterOrEqual(a: IndexedSeq[Cell], v: Long) = {
      def recurse(low: Int, high: Int,lastgood:Int=0):Int = (low + high) / 2 match {
        case _ if high < low => lastgood
        case mid if a(mid).time > v => recurse(low, mid - 1, mid)
        case mid if a(mid).time < v => recurse(mid + 1, high, mid)
        case mid => mid
      }
      recurse(0, a.size - 1)
  }



  /**
    * Combine all given series cells with the current one  
    * @param that An cell's iterable
    * @return the resulting series
    */

  def <<<[X<:Cell](that:Iterable[X]):Series[C] = {
    var ns = new Series[C](name, tm, alias)
    var lthis = this.toStream
    var lthat = that.toStream
    var done=false
    while (!done) {
      (lthis.headOption, lthat.headOption) match {
        case (Some(cthis),Some(cthat)) =>
          if (cthis.time < cthat.time)  { ns = ns << builder.buildFrom(cthis.time, cthis) ; lthis = lthis.tail}
          else                          { ns = ns << builder.buildFrom(cthat.time, cthat) ; lthat=lthat.tail }
        case (Some(cthis), None) =>       ns = ns << builder.buildFrom(cthis.time, cthis) ; lthis = lthis.tail
        case (None, Some(cthat)) =>       ns = ns << builder.buildFrom(cthat.time, cthat) ; lthat=lthat.tail
        case (None,None) => done=true
      }
    }
    ns
  }
  
  /**
    * Combine all given tuples iterable with the current series, any number's type is accepted.
    * @param that An tuple2 iterable
    * @return the resulting series
    */
  def <<<[TN<%Number, CN<%Number] (that:Iterable[Tuple2[TN,CN]]): Series[C] = {
    this <<< that.map {case (time:TN, value:CN) => builder.buildFrom(time.longValue, value.doubleValue) }
  }
  

  /**
    * Combine all given list of Cell's iterable with the current series
    * @param that A list of cell's iterables
    * @return the resulting series
    */
  def <<< [X<:Cell] (that:List[Iterable[X]]):Series[C] = {
    var ns = new Series[C](name, tm, alias)
    var current = that
    var moretodo=true
    do {
      val cells   = (current map {_.headOption} filter {_.isDefined} map {_.get})
      if (cells.isEmpty) moretodo = false
      else {
        val minTime = (cells map {_.time}).min
        val syncCells = cells filter {_.time == minTime}
        ns <<<= syncCells
        current = current map {s => s.headOption match {case Some(head) if (head.time==minTime) => s.tail case _ => s} }
      }
    } while(moretodo)
    ns
  }


  // mathematic operations between a serie and a numeric value
  private def generic(op2:(Double)=>Double) = {
    ( (new Series[CalcCell](name, tm, alias) (CellBuilder.calcCellBuilder)) /: this) {
      (ns, cell)  =>  ns << CellBuilder.calcCellBuilder.buildFrom(cell.time, op2(cell.value))
    }
  }
  def +[N<%Number](that:N) = generic( _ + that.doubleValue)
  def -[N<%Number](that:N) = generic( _ - that.doubleValue)
  def *[N<%Number](that:N) = generic( _ * that.doubleValue)
  def /[N<%Number](that:N) = generic( _ / that.doubleValue)
  def unary_- = generic(-_)
  def unary_+ = generic(+_)
  
  /* PREVIOUS IMPLEMENTATION
  // mathematic operations between series
  // Please keep in mind that operations are only done for common timestamp between the two series
  //   other values are ignored !
  private def generic(that:Iterable[Cell], op:(Double,Double)=>Double):Series[CalcCell] = {
    val commonTimes = (this.toStream map {_.time}) intersect (that.toStream map {_.time})
    var commonLeft  = this.toStream
    var commonRight = that.toStream
    var newseries   = new Series[CalcCell](name, tm)
    for(time<-commonTimes) {
      commonLeft  = commonLeft  dropWhile {_.time < time}
      commonRight = commonRight dropWhile {_.time < time}
      newseries <<= CellBuilder.calcCellBuilder.buildFrom(time, op(commonLeft.head.value,commonRight.head.value))
    }
    newseries
  }
  */
  /* NEW IMPLEMENTATION */
  // mathematic operations between series
  // Warning : only left operand series (LOS) times are kept, and when a right operand series (ROS) time
  // is missing, we took the value for the previous time, or if not available the first ROS value (the head) 
  // This previous is better than the previous one, but we must make the behavior a "parameter"
  private def generic(that:Iterable[Cell], op:(Double,Double)=>Double):Series[CalcCell] = {
    var currentROS=that.toList
    var rosCell=currentROS.head
    var newseries   = new Series[CalcCell](name, tm, alias)
    
    for(losCell <- this) {
      val time = losCell.time
      currentROS.span {_.time < time} match {
        case (Nil,Nil) =>
        case (Nil,after) =>
        case (before,Nil) => 
            rosCell = before.last
            currentROS = Nil
        case (before,after) if (after.head.time==time) =>
            rosCell=after.head
            currentROS = after
        case (before,after) => 
            rosCell = before.last
            currentROS = rosCell::after
      }
      val nv   = op(losCell.value, rosCell.value)
      newseries <<= CellBuilder.calcCellBuilder.buildFrom(time, nv)
    }
    
    newseries
  }
  
  def +(that:Iterable[Cell]) = generic(that, _ + _)
  def -(that:Iterable[Cell]) = generic(that, _ - _)
  def *(that:Iterable[Cell]) = generic(that, _ * _)
  def /(that:Iterable[Cell]) = generic(that, _ / _)
  
  
  override def apply(index:Int):C = backend(index)
  override def length = backend.length
  override def last = backend(length-1)
  override def head = backend.head
  override def headOption = backend.headOption

  /**
    * Sample this series into a new one
    * @param period new sampling period 
    * @return the new sampled series keep the same cell type as the origin
    */
  def sample(newtm:TimeModel) = {
    //new Series[StatCell](name, newtm, alias) <<< this
    new Series[C](name, newtm, alias) <<< this
  }

  /**
    * Sample this series into a new one using StatCell !!
    * @param period new sampling period 
    * @return the new sampled series
    */
  def statSample(newtm:TimeModel) = {
    new Series[StatCell](name, newtm, alias) <<< this
  }

  /**
    * Convert series cells to a new cell type 
    * @return the new series using the new cell type
    */
  def toSeries[Z<:Cell]()(implicit builder:CellBuilder[Z]) = {
    new Series[Z](name, tm, alias) <<< this
  }
  
  
  /**
    * returns a sub-part of the current series, only take cells which belongs to the specified timerange
    * @param fromTime 
    * @param toTime 
    * @return sub timerange series  
    */
  def take(fromTime:Long, toTime:Long) = new Series[C](name, tm, alias, backend dropWhile {_.time < fromTime} takeWhile {_.time <= toTime})

  /**
    * returns a sub-part of the current series, only the first "howlong" time
    * @param howlong how long time to take at the start of the series
    * @return sub timerange series  
    */  
  def take(howlong:Duration) = {
    val to = head.time+howlong.value
    new Series[C](name, tm, alias, backend takeWhile {_.time <= to})
  }
  
  /**
    * returns a sub-part of the current series, drop the first "howlong" time
    * @param howlong how long time to drop at the start of the series
    * @return sub timerange series  
    */  
  def drop(howlong:Duration) = {
    val from = head.time+howlong.value
    new Series[C](name, tm, alias, backend dropWhile {_.time < from})
  }

  /**
    * returns a sub-part of the current series, only the last "howlong" time
    * @param howlong how long time to keep at the end of the series
    * @return sub timerange series  
    */  
  def takeRight(howlong:Duration) = {
    val to = last.time-howlong.value
    new Series[C](name, tm, alias, backend dropWhile {_.time < to})
  }
  
  /**
    * returns a sub-part of the current series, drop the last "howlong" time
    * @param howlong how long time to drop at the end of the series
    * @return sub timerange series  
    */  
  def dropRight(howlong:Duration) = {
    val from = last.time+howlong.value
    new Series[C](name, tm, alias, backend takeWhile {_.time <= from})
  }

  
  /**
    * Build a new series from a given value contained into each cell
    * @param val2get the cell value extractor, for example : _.max for a StatCell
    * @return news series made of extracted cells values  
    */
  def extract(val2get:C=>Double = (x:C)=>x.value) = {
    ((new Series[CalcCell](name, tm, alias) (CellBuilder.calcCellBuilder)) /: this)   {
           (series, cell) =>  series << CellBuilder.calcCellBuilder.buildFrom(cell.time, val2get(cell)) }
  }

  /**
    * Cells values differences, the new series size will be the current one minus one.
    * @return delta series  
    */
  def delta : Series[CalcCell] = {
    var result = new Series[CalcCell](name, tm, alias)(CellBuilder.calcCellBuilder)
    this.backend.sliding(2,1) foreach {
        case Vector(c1,c2) if (c2.value-c1.value >= 0) => result <<= CellBuilder.calcCellBuilder.buildFrom(c2.time, c2.value-c1.value)
        case _ =>
    }
    result
  }
  
  /**
   * Compute cumulated ellapsed time spend by consecutive cells that satisfy the given test
   * @return Cumulated ellapsed time 
   */
  def howlongFor(condition:(Double)=>Boolean):Long = {
    var duration=0L
    for (Vector(c1,c2) <- backend.sliding(2,1)) {
      if (condition(c1.value) && condition(c2.value)) duration += c2.time - c1.time
    }
    duration
  }
  
  /**
    * Remove consecutive identical cell values, the oldest is the one kept.
    * @return compacted series  
    */
  def compact() : Series[CalcCell] = {
    (new Series[CalcCell](name, tm, alias)(CellBuilder.calcCellBuilder) /: this) { (series, thisCell) => series.lastOption match { 
        case Some(thatCell) if (thatCell.value == thisCell.value) => series
        case Some(_)|None=> series << thisCell
      }
    }
  }
  
  /**
    * Cumulates values into a new series. It computes a new series with cellN = cell(N-1)+cellN
    * @return cumulate series  
    */
  def cumulate() : Series[CalcCell] = {
     var result = new Series[CalcCell](name, tm, alias)(CellBuilder.calcCellBuilder)
     var sum=0d
     for(cell <- this.backend) {
       sum += cell.value
       result = result << (cell.time -> sum)
     }
     result
  }

  /**
    * get values fiels of all cells
    * @return values iterable  
    */
  def values : Iterable[Double] = this map {_.value}
  
  /**
    * get time of all cells
    * @return times iterable  
    */
  def times : Iterable[Long] = this map {_.time}

  /**
    * Rename a series
    * @param newName the new series's name
    * @return the renamed series  
    */
  def rename(newName:String) = new Series[C](newName, tm, alias, backend)
  
  /**
    * Re-alias a series (alias is a pretty name given to the series)
    * @param newAlias the new series's alias
    * @return the re-aliased series  
    */
  def realias(newAlias:String=name) = new Series[C](name, tm, newAlias, backend)

  /**
    * Sync series start time with the given series
    * @param series the series to sync time with
    * @return the synchronized series
    */
  def sync(that:Series[Cell]):Series[C] = if (!that.isEmpty) sync(that.head.time) else this
  
  /**
    * Sync series start time with the given time
    * @param newStartTime new series start time
    * @return the synchronized series
    */
  def sync(newStartTime:Long):Series[C] = {
    val deltaTime=newStartTime - head.time
    val newbackend = backend map {cell =>builder.buildFrom(cell.time + deltaTime, cell.value)}
    new Series[C](name, tm, alias, newbackend)
  }
  
  /**
    * makes the series starts with zero value by removing to all cells the first cell value
    * @return the zero based series
    */
  def zeroBased() = if (!isEmpty) this - head.value else this
  
  /**
   * Calculate a rate series from the current series
   * @param ratePeriod rate chosen periodicity, if time is ms, then 1000 will mean 1s
   * @return the rate series
   */
  def toRate(ratePeriod:Long=1000):Series[CalcCell] = {
    var result = new Series[CalcCell](name, tm, alias)(CellBuilder.calcCellBuilder)
    this.backend.sliding(2,1) foreach {
        case Vector(c1,c2) => 
          result <<= CellBuilder.calcCellBuilder.buildFrom(c2.time, c2.value*ratePeriod/(c2.time-c1.time))
        case _ =>
    }
    result
  }
}



object Series {
  def apply[X<:Cell](name:String)(implicit builder:CellBuilder[X]):Series[X] = new Series(name, 1, name)
  def apply[X<:Cell](name:String, alias:String)(implicit builder:CellBuilder[X]):Series[X] = new Series(name, 1, alias)
  def apply[X<:Cell](name:String, timeModel:TimeModel)(implicit builder:CellBuilder[X]):Series[X] = new Series(name, timeModel, name)
  def apply[X<:Cell](name:String, timeModel:TimeModel, alias:String)(implicit builder:CellBuilder[X]):Series[X] = new Series(name, timeModel, alias)  
  def apply[X<:Cell](name:String, series:Series[Cell])(implicit builder:CellBuilder[X]):Series[X] = new Series(name, series.tm, name) <<< series
  def apply[X<:Cell](name:String, alias:String, series:Series[Cell])(implicit builder:CellBuilder[X]):Series[X] = new Series(name, series.tm, alias) <<< series
  def apply[X<:Cell](series:Series[Cell])(implicit builder:CellBuilder[X]):Series[X] = new Series(series.name, series.tm, series.alias) <<< series 
}




